package org.aplikator.server.persistence.empiredb;

import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.io.StringReader;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.sql.DataSource;

import org.apache.empire.data.DataType;
import org.apache.empire.db.DBBlobData;
import org.apache.empire.db.DBCmpType;
import org.apache.empire.db.DBColumn;
import org.apache.empire.db.DBColumnExpr;
import org.apache.empire.db.DBCommand;
import org.apache.empire.db.DBDatabase;
import org.apache.empire.db.DBDatabaseDriver;
import org.apache.empire.db.DBJoinType;
import org.apache.empire.db.DBReader;
import org.apache.empire.db.DBRecord;
import org.apache.empire.db.DBSQLScript;
import org.apache.empire.db.DBTable;
import org.apache.empire.db.DBTableColumn;
import org.apache.empire.db.expr.compare.DBCompareAndOrExpr;
import org.apache.empire.db.expr.compare.DBCompareColExpr;
import org.apache.empire.db.expr.compare.DBCompareExpr;
import org.apache.empire.db.h2.DBDatabaseDriverH2;
import org.apache.empire.db.hsql.DBDatabaseDriverHSql;
import org.apache.empire.db.mysql.DBDatabaseDriverMySQL;
import org.apache.empire.db.oracle.DBDatabaseDriverOracle;
import org.apache.empire.db.postgresql.DBDatabaseDriverPostgreSQL;
import org.apache.empire.db.sqlserver.DBDatabaseDriverMSSQL;
import org.aplikator.client.data.PrimaryKey;
import org.aplikator.client.data.Record;
import org.aplikator.client.descriptor.ViewDTO;
import org.aplikator.client.descriptor.PropertyDTO;
import org.aplikator.client.widgets.BinaryFieldWidget;
import org.aplikator.server.DescriptorRegistry;
import org.aplikator.server.ImageResizer;
import org.aplikator.server.descriptor.View;
import org.aplikator.server.descriptor.Collection;
import org.aplikator.server.descriptor.Entity;
import org.aplikator.server.descriptor.Property;
import org.aplikator.server.descriptor.Reference;
import org.aplikator.server.persistence.Persister;
import org.aplikator.server.persistence.Transaction;
import org.aplikator.server.query.QueryCombineExpression;
import org.aplikator.server.query.QueryCombineOperator;
import org.aplikator.server.query.QueryCompareExpression;
import org.aplikator.server.query.QueryCompareOperator;
import org.aplikator.server.query.QueryExpression;
import org.aplikator.server.util.Configurator;
import org.aplikator.shared.data.BinaryData;

import com.google.common.io.ByteStreams;

public class EmpireDbPersister implements Persister {

    private static final Logger LOG = Logger.getLogger(EmpireDbPersister.class.getName());

    public EmpireDbPersister() {
        initDataSourceName();
    }

    private DBDatabase db = new EmpireDatabase();

    DataSource dataSource = null;
    DBDatabaseDriver driver = null;
    Connection pconn = null;
    private String dataSourceName = null;
    // Map<Integer, PersisterTriggers> registeredtriggers = new HashMap<Integer,
    // PersisterTriggers>();

    Map<Entity, DBTable> tables = new HashMap<Entity, DBTable>();
    Map<Property<? extends Serializable>, DBTableColumn> columns = new HashMap<Property<? extends Serializable>, DBTableColumn>();
    Map<Property<BinaryData>, BlobColumns> blobs = new HashMap<Property<BinaryData>, BlobColumns>();

    public void registerEntity(Entity entity, String tableName) {
        if (tables.containsKey(entity)) {
            throw new IllegalStateException("Duplicate entity registration:" + entity + ", " + tableName);
        }
        DBTable table = new DBTable(tableName, db);
        tables.put(entity, table);
    }

    public void registerPrimaryKey(Entity entity, Property<Integer> property, String primaryKeyName) {
        DBTable table = tables.get(entity);
        if (table == null) {
            throw new IllegalStateException("Cannot register primaryKey " + property + " - entity " + entity + " is not yet registered.");
        }
        DBTableColumn primaryKeycolumn = table.addColumn(primaryKeyName, DataType.AUTOINC, 0, true, primaryKeyName + "_SEQUENCE");
        table.setPrimaryKey(primaryKeycolumn);
        columns.put(property, primaryKeycolumn);
    }

    public void registerTimestamp(Entity entity, Property<Date> property, String timestampName) {
        DBTable table = tables.get(entity);
        if (table == null) {
            throw new IllegalStateException("Cannot register timestamp " + property + " - entity " + entity + " is not yet registered.");
        }
        DBTableColumn timestampColumn = table.addColumn(timestampName, DataType.DATETIME, 0, false);
        table.setTimestampColumn(timestampColumn);
        columns.put(property, timestampColumn);
    }

    @SuppressWarnings("unchecked")
    public <T extends Serializable> void registerProperty(Entity entity, Property<T> property, String columnName) {
        DBTableColumn column = null;
        Class<T> type = property.getType();
        DBTable table = tables.get(entity);
        if (table == null) {
            throw new IllegalStateException("Cannot register property " + property + " - entity " + entity + " is not yet registered.");
        }
        if (type.equals(Boolean.class)) {
            column = table.addColumn(columnName, mapType(property), property.getSize(), property.isRequired(), false);
        } else if (type.equals(BinaryData.class)) {// TODO check if it should
                                                   // use addBlob?
            BlobColumns blob = new BlobColumns();
            blob.column = table.addColumn(columnName, DataType.BLOB, 0, false);
            blob.mediumColumn = table.addColumn(columnName + "_MEDIUM", DataType.BLOB, 0, false);
            blob.fullColumn = table.addColumn(columnName + "_FULL", DataType.BLOB, 0, false);
            blob.propsColumn = table.addColumn(columnName + "_PROPS", DataType.CLOB, 0, false);
            blobs.put((Property<BinaryData>) property, blob);
            // column = blob.propsColumn;
        } else {
            column = table.addColumn(columnName, mapType(property), property.getSize(), property.isRequired());
        }
        if (column != null) {
            columns.put(property, column);
        }
    }

    public void registerReference(Entity entity, Property<Integer> property, String columnName, Entity referredEntity) {
        DBTable table = tables.get(entity);
        if (table == null) {
            throw new IllegalStateException("Cannot register property " + property + " - entity " + entity + " is not yet registered.");
        }
        DBTableColumn columnMany = table.addColumn(columnName, DataType.INTEGER, 0, false);
        DBTableColumn columnOne = columns.get(referredEntity.getPrimaryKey());
        if (columnOne == null) {
            throw new IllegalStateException("Cannot register reference " + property + " - target entity " + referredEntity + " is not yet registered.");
        }
        db.addRelation(columnMany.referenceOn(columnOne));

        columns.put(property, columnMany);
    }

    public void registerReverseCollection(Collection<? extends Entity> many, Reference<? extends Entity> one) {
        DBTableColumn columnOne = columns.get(one);
        if (columnOne == null) {
            throw new IllegalStateException("Cannot register reverse collection " + many + " - reference " + one + " is not yet registered.");
        }
        columns.put(many, columnOne);
    }

    public void registerIndex(Entity entity, String name, boolean unique, Property<? extends Serializable>... properties) {
        DBTable table = tables.get(entity);
        if (table == null) {
            throw new IllegalStateException("Cannot register index " + name + " - entity " + entity + " is not yet registered.");
        }
        DBColumn[] cols = new DBColumn[properties.length];
        for (int i = 0; i < properties.length; i++) {
            DBTableColumn col = columns.get(properties[i]);
            if (col == null) {
                throw new IllegalStateException("Cannot register index " + name + " - property " + properties[i] + " is not yet registered.");
            }
            cols[i] = col;
        }
        table.addIndex(name, unique, cols);
    }

    public void registerEntityClone(Entity original, Entity clone) {
        tables.put(clone, (DBTable) (tables.get(original)).clone());

    }

    public <T extends Serializable> void registerPropertyClone(Property<T> original, Property<T> clone) {
        columns.put(clone, columns.get(original));
    }

    private <T extends Serializable> DataType mapType(Property<T> property) {

        Class<T> aplikatorType = property.getType();
        if (aplikatorType.equals(String.class)) {
            if (property.getSize() > 0) {
                return DataType.TEXT;
            } else {
                return DataType.CLOB;
            }
        }
        if (aplikatorType.equals(Boolean.class)) {
            return DataType.BOOL;
        }
        if (aplikatorType.equals(BigDecimal.class)) {
            return DataType.DECIMAL;
        }
        if (aplikatorType.equals(Integer.class)) {
            return DataType.INTEGER;
        }
        if (aplikatorType.equals(Date.class)) {
            return DataType.DATETIME;
        }
        throw new IllegalStateException("Unsupported property type:" + aplikatorType);
    }

    private void initDataSourceName() {
        dataSourceName = Configurator.get().getConfig().getString(Configurator.DATASOURCE);
    }

    /*
     * (non-Javadoc)
     *
     * @see org.aplikator.server.Persister#getJDBCConnection()
     */
    @SuppressWarnings("serial")
    public Connection getJDBCConnection() {
        if (dataSource == null) {// lazy setup of datasource
            Context ctx = null;
            try {
                ctx = new InitialContext();
                dataSource = (DataSource) ctx.lookup(dataSourceName);
            } catch (NamingException e) {
                throw new IllegalStateException("Error getting datasource " + dataSourceName + " from JNDI.", e);
            }
        }

        if (driver == null) {
            Connection conn = null;
            try {
                conn = dataSource.getConnection();
                String dbName = conn.getMetaData().getDatabaseProductName();
                if (dbName.contains("PostgreSQL")) {
                    driver = new DBDatabaseDriverPostgreSQL(){//temporary fix of empire-db clob bug

                        @Override
                        public Object getResultValue(ResultSet rset, int columnIndex, DataType dataType) throws SQLException {
                            if (dataType == DataType.CLOB){
                                return rset.getString(columnIndex);
                            }else if (dataType == DataType.BLOB){
                                InputStream is =  rset.getBinaryStream(columnIndex);
                                try{
                                    return ByteStreams.toByteArray(is);
                                }catch (IOException e) {
                                    LOG.log(Level.SEVERE, "",e);
                                    return new byte[]{};
                                }
                                finally{
                                    if (is != null)
                                        try {
                                            is.close();
                                        } catch (IOException e) {}
                                }
                            }
                            else
                                return super.getResultValue(rset, columnIndex, dataType);
                        }

                    };
                } else if (dbName.contains("Oracle")) {
                    driver = new DBDatabaseDriverOracle();
                } else if (dbName.contains("MySQL")) {
                    driver = new DBDatabaseDriverMySQL();
                } else if (dbName.contains("MS-SQL")) {
                    driver = new DBDatabaseDriverMSSQL();
                } else if (dbName.contains("HSQL")) {
                    driver = new DBDatabaseDriverHSql();
                } else if (dbName.contains("H2")) {
                    driver = new DBDatabaseDriverH2();
                } else {
                    throw new IllegalStateException("Unsupported database:" + dbName);
                }

            } catch (SQLException e) {
                throw new RuntimeException(e);
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (SQLException e) {
                    }
                }
            }
        }

        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);
            db.open(driver, conn);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        return conn;
    }

    private class EmpireDbTransaction implements Transaction {
        private final Connection conn;

        private EmpireDbTransaction() {
            conn = getJDBCConnection();
        }

    }

    public Transaction beginTransaction() {
        return new EmpireDbTransaction();
    }

    public void commitTransaction(Transaction tx) {
        db.commit(((EmpireDbTransaction) tx).conn);
    }

    public void close(Transaction tx) {
        Connection conn = ((EmpireDbTransaction) tx).conn;
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                LOG.log(Level.SEVERE, "Error in close connection", e);
            }
        }
    }

    public String generateDDL(boolean updateDB) {
        Connection conn = getJDBCConnection();
        StringBuilder retval = new StringBuilder();
        DBSQLScript script = new DBSQLScript();
        db.getCreateDDLScript(driver, script);
        if (updateDB) {
            try {
                script.run(driver, conn, false);
                db.commit(conn);
                retval.append("Script executed sucessfuly.");
                retval.append("\n\n");
            } catch (Exception e) {
                retval.append("Script not executed:" + e);
                retval.append("\n\n");
            } finally {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }
        try {
            conn.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        // return DDL script
        retval.append(script.toString());
        return retval.toString();

    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.aplikator.server.Persister#getPage(org.aplikator.client.descriptor
     * .ViewDTO, org.aplikator.client.descriptor.PropertyDTO,
     * org.aplikator.client.data.PrimaryKeyDTO, int, int)
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public List<Record> getPage(ViewDTO vd, PropertyDTO<Integer> ownerProperty, PrimaryKey ownerPrimaryKey, int pageOffset, int pageSize, org.aplikator.server.Context ctx) {
        DBReader reader = null;
        Connection conn = null;
        List<Record> retval = new ArrayList<Record>();
        try {
            conn = getJDBCConnection();
            // Define the query
            View view = (View) DescriptorRegistry.get().getDescriptionItem(vd.getId());
            DBCommand cmd = db.createCommand();
            // Select required columns
            DBColumnExpr[] tableColumns = new DBColumnExpr[view.getProperties().size() + 1];
            tableColumns[0] = columns.get(view.getEntity().getPrimaryKey());
            for (int i = 1; i <= view.getProperties().size(); i++) {
                Property<? extends Serializable> property = view.getProperties().get(i - 1);
                tableColumns[i] = columns.get(property);
                if (property.getRefferedThrough() != null) {
                    cmd.join(columns.get(property.getRefferedThrough()), columns.get(property.getRefferedThrough().referredEntity.getPrimaryKey()), DBJoinType.LEFT);
                    // cmd.join(BK.AUTHOR_ID, AUT.AUTHOR_ID);
                }
            }
            cmd.select(tableColumns);

            if (ownerProperty != null) {
                DBColumnExpr ownerColumn = columns.get(((Property<Integer>) DescriptorRegistry.get().getDescriptionItem(ownerProperty.getId())));
                cmd.where(ownerColumn.is(ownerPrimaryKey.getId()));
            }

            DBCompareExpr where = convertQueryExpression(view.getQueryGenerator().createWhere(vd.getQueryParameters(), ctx));
            if (where != null) {
                addWhere(cmd, where);
            }

            if (vd.getSortProperty() != null) {
                Property<? extends Serializable> sortProp = (Property<? extends Serializable>) DescriptorRegistry.get().getDescriptionItem(vd.getSortProperty().getId());
                /*
                 * if (vd.getSearchString() != null &&
                 * !vd.getSearchString().equals("")) { addWhere(cmd,
                 * sortProp.column.isMoreOrEqual(vd.getSearchString())); }
                 */
                cmd.orderBy(columns.get(sortProp), !vd.isSortAscending());
            }
            cmd.orderBy(columns.get(view.getEntity().getPrimaryKey()));

            // Query Records and print output
            reader = new DBReader();

            // Open Reader
            LOG.fine("GETPAGE Query: " + cmd.getSelect());
            reader.open(cmd, conn);

            if (pageSize > 0) {
                reader.skipRows(pageOffset);
            }
            int recCounter = 0;
            while (reader.moveNext()) {
                Record rec = new Record(new PrimaryKey(vd.getEntity().getId(), reader.getInt(columns.get(view.getEntity().getPrimaryKey()))));
                for (int i = 0; i < view.getProperties().size(); i++) {
                    Property p = view.getProperties().get(i);
                    if (Integer.class.equals(p.getType())){
                        p.setValue(rec,  reader.getInt(columns.get(p)));
                    }else{
                        p.setValue(rec, (Serializable) reader.getValue(columns.get(p)));
                    }
                }
                rec.resetDirtyFlags();
                retval.add(rec);
                recCounter++;
                if (pageSize > 0 && recCounter >= pageSize) {
                    break;
                }

            }

        } catch (Throwable th) {
            LOG.log(Level.SEVERE, "Error in GET PAGE", th);
        } finally {
            if (reader != null) {
                reader.close();
            }
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException ex) {
            }
        }
        return retval;
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.aplikator.server.Persister#getRecordCount(org.aplikator.client.descriptor
     * .ViewDTO, org.aplikator.client.descriptor.PropertyDTO,
     * org.aplikator.client.data.PrimaryKeyDTO, boolean)
     */
    public int getRecordCount(ViewDTO vd, PropertyDTO<Integer> ownerProperty, PrimaryKey ownerPrimaryKey, boolean search, org.aplikator.server.Context ctx) {
        DBReader reader = null;
        Connection conn = null;
        try {
            conn = getJDBCConnection();
            // Define the query
            View view = (View) DescriptorRegistry.get().getDescriptionItem(vd.getId());
            DBCommand cmd = db.createCommand();
            // Select count(*)
            cmd.select(tables.get(view.getEntity()).count());

            if (ownerProperty != null) {
                // TODO Doplnit sloupce kvuli join a where
                @SuppressWarnings("unchecked")
                DBColumnExpr ownerColumn = columns.get(((Property<Integer>) DescriptorRegistry.get().getDescriptionItem(ownerProperty.getId())));
                cmd.where(ownerColumn.is(ownerPrimaryKey.getId()));
            }

            DBCompareExpr where = convertQueryExpression(view.getQueryGenerator().createWhere(vd.getQueryParameters(), ctx));
            if (where != null) {
                addWhere(cmd, where);
            }

            if (search && vd.getSortProperty() != null) {
                @SuppressWarnings("unchecked")
                Property<? extends Serializable> sortProp = (Property<? extends Serializable>) DescriptorRegistry.get().getDescriptionItem(vd.getSortProperty().getId());
                DBTableColumn sortCol = columns.get(sortProp);
                if (vd.getSearchString() != null && !vd.getSearchString().equals("")) {
                    if (vd.isSortAscending()) {
                        addWhere(cmd, sortCol.isSmallerThan(vd.getSearchString()));
                    } else {
                        addWhere(cmd, sortCol.isGreaterThan(vd.getSearchString()));
                    }
                }
                // cmd.orderBy(sortProp.column, !vd.isSortAscending());
            }

            // Query Records and print output
            reader = new DBReader();

            // Open Reader
            LOG.fine("getRecord Count Query: " + cmd.getSelect());
            reader.open(cmd, conn);
            if (reader.moveNext()) {
                return reader.getInt(0);
            }

        } catch (Throwable th) {
            LOG.log(Level.SEVERE, "Error in getRecordCount", th);
        } finally {
            if (reader != null) {
                reader.close();
            }
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException ex) {
            }
        }
        return 0;
    }

    private void addWhere(DBCommand cmd, DBCompareExpr where) {
        List<DBCompareExpr> templist = Arrays.asList(where);
        cmd.addWhereConstraints(templist);
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    public Record getRecord(ViewDTO viewDTO, PrimaryKey primaryKey, org.aplikator.server.Context ctx) {
        Connection conn = null;
        Record retval = null;
        DBReader reader = null;
        try {
            conn = getJDBCConnection();
            View view = (View) DescriptorRegistry.get().getDescriptionItem(viewDTO.getId());

            DBCommand cmd = db.createCommand();
            // Select required columns
            DBColumnExpr[] tableColumns = new DBColumnExpr[view.getForm(ctx).getProperties().size() + 1];
            tableColumns[0] = columns.get(view.getEntity().getPrimaryKey());
            for (int i = 1; i <= view.getForm(ctx).getProperties().size(); i++) {
                Property<? extends Serializable> property = view.getForm(ctx).getProperties().get(i - 1);
                tableColumns[i] = columns.get(property);
                if (property.getRefferedThrough() != null) {
                    cmd.join(columns.get(property.getRefferedThrough()), columns.get(property.getRefferedThrough().referredEntity.getPrimaryKey()), DBJoinType.LEFT);
                }
            }
            cmd.select(tableColumns);
            cmd.where(columns.get(view.getEntity().getPrimaryKey()).is(primaryKey.getId()));

            reader = new DBReader();
            LOG.fine("GET RECORD Query: " + cmd.getSelect());
            reader.open(cmd, conn);

            if (reader.moveNext()) {
                retval = new Record(new PrimaryKey(viewDTO.getEntity().getId(), reader.getInt(columns.get(view.getEntity().getPrimaryKey()))));
                for (int i = 0; i < view.getForm(ctx).getProperties().size(); i++) {
                    Property p = view.getForm(ctx).getProperties().get(i);
                    if (Integer.class.equals(p.getType())){
                        p.setValue(retval,  reader.getInt(columns.get(p)));
                    }else{
                        p.setValue(retval, (Serializable) reader.getValue(columns.get(p)));
                    }
                }
                retval.resetDirtyFlags();
            } else {
                throw new IllegalStateException("Record not found:" + primaryKey);
            }

        } catch (Throwable th) {
            LOG.log(Level.SEVERE, "Error in getRecord ", th);
        } finally {
            try {
                if (reader != null) {
                    reader.close();
                }
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException ex) {
            }
        }
        return retval;
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.aplikator.server.Persister#updateRecord(org.aplikator.client.descriptor
     * .ViewDTO, org.aplikator.client.data.RecordDTO, java.sql.Connection)
     */
    public Record updateRecord(String entityId, Record record, Transaction tx, org.aplikator.server.Context ctx) {
        Entity entity = (Entity) DescriptorRegistry.get().getDescriptionItem(entityId);

        Connection conn = ((EmpireDbTransaction) tx).conn;
        DBRecord rec = new DBRecord();
        if (record.getPrimaryKey().getId() == -1) {
            rec.create(tables.get(entity), conn);
            if (record.getOwnerPropertyId() != null) {
                @SuppressWarnings("unchecked")
                DBTableColumn ownerColumn = columns.get(((Property<Integer>) DescriptorRegistry.get().getDescriptionItem(record.getOwnerPropertyId())));
                rec.setValue(ownerColumn, record.getOwnerPrimaryKey().getId());
            }
        } else {
            rec.read(tables.get(entity), record.getPrimaryKey().getId(), conn);
        }
        /*
         * for (int i = 0; i < view.getForm(ctx).getProperties().size(); i++) {
         * Property<? extends Serializable> p =
         * view.getForm(ctx).getProperties().get(i); if (p.getRefferedThrough()
         * != null) { continue; } if (record.isDirty(p.getId())){
         * rec.setValue(columns.get(p), p.getValue(record)); } }
         */

        for (String id : record.getProperties()) {
            @SuppressWarnings("unchecked")
            Property<? extends Serializable> p = (Property<? extends Serializable>) DescriptorRegistry.get().getDescriptionItem(id);
            if (p.getRefferedThrough() != null) {
                continue;
            }
            if (record.isDirty(p.getId())) {
                rec.setValue(columns.get(p), p.getValue(record));
            }
        }

        rec.update(conn);
        if (record.getPrimaryKey().getId() == -1) {
            record.getPrimaryKey().setId(rec.getInt(columns.get(entity.getPrimaryKey())));
        }

        return record;
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.aplikator.server.Persister#deleteRecord(org.aplikator.client.descriptor
     * .ViewDTO, int, java.sql.Connection)
     */
    @SuppressWarnings("rawtypes")
    public void deleteRecord(String entityId, int primaryKey, Transaction tx, org.aplikator.server.Context ctx) {
        Entity entity = (Entity) DescriptorRegistry.get().getDescriptionItem(entityId);
        Connection conn = ((EmpireDbTransaction) tx).conn;

        //find and delete any nested collections
        //TODO: delete through command API, to call triggers on collections and to propagate further nesting...
        for (Property p:entity.getProperties()){
            if (p instanceof Collection){
                Collection c = (Collection) p;
                Entity nestedEntity = c.referredEntity;
                DBCommand cmd = db.createCommand();
                // Select pk column of the items to delete
                DBColumnExpr[] tableColumns = new DBColumnExpr[1];
                tableColumns[0] = columns.get(nestedEntity.getPrimaryKey());

                cmd.select(tableColumns);
                cmd.where(columns.get(c).is(primaryKey));//where many column = ownerPK

                DBReader reader = new DBReader();
                LOG.fine("GET SUBRECORD for delete Query: " + cmd.getSelect());
                reader.open(cmd, conn);
                List<Integer> subpks = new ArrayList<Integer>();
                while (reader.moveNext()) {
                    subpks.add( reader.getInt(columns.get(nestedEntity.getPrimaryKey())));//read the list of pks to delete
                }
                if (subpks.size()>0){
                    tables.get(nestedEntity).deleteRecord(subpks.toArray(), conn);//delete the list
                }
            }
        }


        tables.get(entity).deleteRecord(primaryKey, conn);
    }

    public BinaryData getBlob(Entity entity, Property<BinaryData> property, int primaryKey, int maxSize) {
        Connection conn = null;
        BlobColumns blob = blobs.get(property);
        BinaryData retval = new BinaryData();
        DBReaderExt reader = null;
        try {
            conn = getJDBCConnection();

            DBCommand cmd = db.createCommand();
            // Select required columns
            DBColumnExpr[] tableColumns = new DBColumnExpr[2];
            tableColumns[0] = blob.propsColumn;
            if (maxSize == BinaryFieldWidget.THUMBNAIL_SIZE_CODE) {// thumbnail
                tableColumns[1] = blob.column;
            } else if (maxSize == BinaryFieldWidget.FULL_SIZE_CODE) {// fullSize
                tableColumns[1] = blob.fullColumn;
            } else if (maxSize == BinaryFieldWidget.MEDIUM_SIZE_CODE) {// mediumSize
                tableColumns[1] = blob.mediumColumn;
            } else {// scaled preview
                tableColumns[1] = blob.fullColumn;
            }
            cmd.select(tableColumns);
            cmd.where(columns.get(entity.getPrimaryKey()).is(primaryKey));

            reader = new DBReaderExt();
            LOG.fine("GET BLOB Query: " + cmd.getSelect());
            reader.open(cmd, conn);

            if (reader.moveNext()) {
                Properties loaderProps = new Properties();
                ResultSet rs = reader.getResultset();
                String propstr = rs.getString(1);
                if (propstr != null) {
                    loaderProps.load(new StringReader(propstr));
                    for (Object key : loaderProps.keySet()) {
                        retval.properties.put(key.toString(), loaderProps.getProperty(key.toString()));
                    }
                }
                if (maxSize <= 0) {// thumbnail, medium, full
                    retval.data = (byte[]) rs.getBytes(2);
                } else {// scaled preview
                    retval.data = ImageResizer.resize((byte[]) rs.getBytes(2), retval.properties.get("filename"), maxSize);
                }

            } else {
                throw new IllegalStateException("Record for blob not found:" + primaryKey);
            }

        } catch (Throwable th) {
            LOG.log(Level.SEVERE, "Error in getBlob ", th);
        } finally {
            try {
                if (reader != null) {
                    reader.close();
                }
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException ex) {
            }
        }
        if (retval.data == null) {
            retval.data = new byte[0];
        }
        return retval;
    }

    private static class DBReaderExt extends DBReader {
        /**
         *
         */
        private static final long serialVersionUID = 1L;

        private ResultSet getResultset() {
            return rset;
        }
    }

    public void updateBlob(Entity entity, Property<BinaryData> property, int primaryKey, String filename, byte[] data) {
        Connection conn = null;
        BlobColumns blob = blobs.get(property);
        try {
            conn = getJDBCConnection();

            DBRecord rec = new DBRecord();

            rec.read(tables.get(entity), primaryKey, conn);

            rec.setValue(blob.fullColumn, new DBBlobData(data));
            byte[] icon = ImageResizer.resize(data, filename, BinaryFieldWidget.THUMBNAIL_SIZE);
            rec.setValue(blob.column, new DBBlobData(icon));
            byte[] preview = ImageResizer.resize(data, filename, BinaryFieldWidget.PREVIEW_SIZE);
            rec.setValue(blob.mediumColumn, new DBBlobData(preview));

            rec.setValue(blob.propsColumn, "filename=" + filename);// TODO other
                                                                   // properties

            rec.update(conn);
            db.commit(conn);
        } catch (Throwable th) {
            LOG.log(Level.SEVERE, "Error while updating Blob:", th);
            throw new RuntimeException("Error while updating Blob:", th);
        } finally {
            try {
                conn.close();
            } catch (SQLException e) {
                LOG.log(Level.SEVERE, "Error while closing connection:", e);
            }
        }
    }

    private static class BlobColumns {
        private DBTableColumn column;
        private DBTableColumn fullColumn;
        private DBTableColumn mediumColumn;
        private DBTableColumn propsColumn;
    }

    private DBCompareExpr convertQueryExpression(QueryExpression sourceExpression) {
        if (sourceExpression instanceof QueryCombineExpression) {
            QueryCombineExpression srcExp = (QueryCombineExpression) sourceExpression;
            return new DBCompareAndOrExpr(convertQueryExpression(srcExp.getLeft()), convertQueryExpression(srcExp.getRight()), srcExp.getOperator().equals(QueryCombineOperator.OR));
        }
        if (sourceExpression instanceof QueryCompareExpression<?>) {
            QueryCompareExpression<? extends Serializable> srcExp = (QueryCompareExpression<?>) sourceExpression;
            return new DBCompareColExpr(columns.get(srcExp.getProperty()), convertCompareOperator(srcExp.getOperator()), srcExp.getValue());
        }
        return null;
    }

    private DBCmpType convertCompareOperator(QueryCompareOperator operator) {
        switch (operator) {
        case IS:
            return DBCmpType.EQUAL;
        case LIKE:
            return DBCmpType.LIKE;
        default:
            return null;
        }
    }

}
